package docsim;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.StringTokenizer;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.log.*;

public class Map1 extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> {
	private static final int ZERO = 0;

	private static final String SPACE = " ";

	private static final String SLASH = "/";

	private  LongWritable docid;
	
	public String inputFile;
	//public static String space = " ";
	private  Text wordToPut = new Text();
	public  HashMap<Integer,String> idDocMap = new HashMap<Integer,String>(); 
    public  HashMap<String,Integer> docIdMap = new HashMap<String,Integer>();

	 	@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, LongWritable> output,
				Reporter reporter) throws IOException {
	 		//long b = System.currentTimeMillis();
	 		//hdfs://localhost:9000/home/deepak/pubmed/PLoS_Biol_2003_Nov_17_1(2)_e53.nxml
	 		//System.out.println("haha!");
	 		//System.out.println(docIdMap.size());
	 		inputFile = inputFile.substring(inputFile.lastIndexOf(SLASH)+1);
	 		//System.out.println(inputFile);
	 		Integer i = docIdMap.get(inputFile);
	 		
			docid = new LongWritable(i);
			
			String content = Preprocessor.preprocessText(value.toString());
			//String content = value.toString();
			String[] words = content.toString().toLowerCase().split(SPACE);
			HashSet<String> setWords = new HashSet<String>();
			for(String word: words)
			{
				setWords.add(word);
			}
			for(String word: setWords)
			{
				if(word.length()!=ZERO)
				{
					wordToPut.set(word);
					output.collect(wordToPut, docid);
				}
			}
			//long a = System.currentTimeMillis();
	        //double id = Math.random();
	        //System.out.println("Map1 Time:"+id+":"+(a-b));
		}
	 	
	 	private void loadDocIdMap(Path docIdsFile) {
            try {
              BufferedReader fis = new BufferedReader(new FileReader(docIdsFile.toString()));
              String pattern = null;
              Integer key;
              String value;
              while ((pattern = fis.readLine()) != null) {
           	    StringTokenizer st1 = new StringTokenizer(pattern);
           	    key = Integer.parseInt(st1.nextToken());
           	    value = st1.nextToken();
           	    idDocMap.put(key, value);
           	    docIdMap.put(value, key);
              }
            } catch (IOException ioe) {
              System.err.println("Caught exception while parsing the cached file '" + idDocMap + "' : " + StringUtils.stringifyException(ioe));
            }
          }
	 	
	   @Override
       public void configure(JobConf job) {
		   //long b = System.currentTimeMillis();
           inputFile = job.get("map.input.file");
           
           if (job.getBoolean("docsim.docids.map", true)) {
               Path[] docidMapFile = null;
               try {
                 docidMapFile = DistributedCache.getLocalCacheFiles(job);
               } catch (IOException ioe) {
                 System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe));
               }
               for (Path docidFile : docidMapFile) {
                 //if(docidFile.getName().matches("docid_map.txt"))
                	loadDocIdMap(docidFile);
               }
             }
            //long a = System.currentTimeMillis();
	        //double id = Math.random();
	        //System.out.println("config2 Time:"+id+":"+(a-b));
         }
}
          